package com.sunzm.flink.datastream.scala.cep

import java.time.Duration

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.Map

object ScalaCepDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    val watermarkStrategy = WatermarkStrategy
      //数据没有乱序，WaterMark不需要设置乱序程度
      .forBoundedOutOfOrderness[LoginEvent](Duration.ofSeconds(0))
      .withTimestampAssigner(new SerializableTimestampAssigner[LoginEvent] {
        override def extractTimestamp(element: LoginEvent, recordTimestamp: Long): Long = {
          //数据里没有时间，就使用系统时间了
          System.currentTimeMillis()
        }
      })

    //注意： 使用CEP需要使用事件时间语义
    val loginEventStream = env.fromElements[LoginEvent](
      LoginEvent("1", "192.168.0.1", "fail"),
      LoginEvent("1", "192.168.0.2", "fail"),
      LoginEvent("1", "192.168.0.3", "fail"),
      LoginEvent("2", "192.168.10,10", "success")
    ).assignTimestampsAndWatermarks(watermarkStrategy)

    val loginFailPattern = Pattern.begin[LoginEvent]("begin")
      .where(_.loginType.equals("fail"))
      .times(2)
      .within(Time.seconds(1))

    val patternStream = CEP.pattern(loginEventStream.keyBy(_.userId), loginFailPattern)

    val loginFailDataStream = patternStream
      .select((pattern: Map[String, Iterable[LoginEvent]]) => {
        val first = pattern.getOrElse("begin", null).iterator.next()

        LoginWarning(first.userId, first.ip, first.loginType)
      })

    loginFailDataStream.print

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  private case class LoginEvent(
                                 userId: String,
                                 ip: String,
                                 loginType: String
                               )

  private case class LoginWarning(
                                   userId: String,
                                   ip: String,
                                   loginType: String
                                 )

}
